package Spark网络日志分析

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.{SparkConf, SparkContext}

object Log {

  var count_valid = 0
  var count_invalid = 0

  def main(args: Array[String]): Unit = {
    // 127.0.0.1 - - [01/Aug/1995:00:00:01 -0400] "GET /images/launch-logo.gif HTTP/1.0" 200 1839
    val conf = new SparkConf()
      .setAppName("log_analysis")
      .setMaster("local[6]")
    val sc = new SparkContext(conf)
    // 设置日志级别
    sc.setLogLevel("ERROR")
    // 转为DataFrame
    val spark = SparkSession.builder()
      .master("local[6]")
      .appName("log")
      .getOrCreate()
    import org.apache.spark.sql.functions._
    import spark.implicits._

    // 正则表达式解析日志行
    val LOG_PATTERN = " (?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)"
    //val data = sc.textFile("src/main/scala/Spark网络日志分析/data/log")
    val data = sc.textFile("G:\\Projects\\IdeaProjects\\Spark_Competition\\dataset\\log.log")

    // 处理
    val clean = data
      .map {
        item =>
          val line = item
            // 1.字段初步拆分
            .replace("[", "\"").replace("]", "\"")
            .split(LOG_PATTERN, -1)

          // 2.转换时间数据
          line(3) = line(3).substring(1, line(3).length - 1)
          val date = parseDate(line(3))

          // 3.字段细分 --- 请求、协议
          // POST /images/launch-logo.gif HTTP/1.0
          line(4) = line(4).substring(1, line(4).length - 1)
          val fields = line(4).split(" ")
          // 222.141.54.75 - - [30/Mar/2015:18:32:26 +0800] "GET / HTTP/1.1" 301 -
          if (fields.length == 3) {
            count_valid += 1
            val method = fields(0)
            val endpoint = fields(1)
            val protocol = fields(2)
            // 转换数据 - =》0
            if ("-".equals(line(6))) {
              line(6) = "0"
            }
            // 4.返回结果
            // "host", "client_id", "user_id", "date_time", "method", "endpoint", "protocol", "response_code", "size"
            (line(0), line(1), line(2), date, method, endpoint, protocol, line(5), line(6))
          } else {
            // 不合法字段返回 “null”
            count_invalid += 1
            (line(0), line(1), line(2), date, "null", "null", "null", line(5), line(6))
          }
      }
    clean.foreach(println(_))
    println(count_valid + "+" + count_invalid + "=" + (count_valid + count_invalid))

    val df_data = clean.toDF("host", "client_id", "user_id", "date_time", "method", "endpoint", "protocol", "response_code", "size")
    df_data.show()

    df_data.select('*)
      .where('method === "null")
      .where('endpoint === "null")
      .where('protocol === "null")
      .show()
    val df = df_data.select('*)
      .where('protocol =!= "null")
    //    df.coalesce(1) // 设置分区为1
    //      .write.mode("overwrite")
    //      .option("header", "true")
    //      .csv("G:\\Projects\\IdeaProjects\\Spark_Competition\\src\\main\\scala\\Spark网络日志分析\\data\\test01")

    // 1.统计Web服务器返回的内容大小，即字段 size ，并计算其平均值、最小值和最大值。
    /*方法一*/
    // 首先调用变换函数 map() 并传入一个只取出字段 size 的匿名函数创建一个新的RDD 。
    // 再调用动作函数 reduce() 和 count() 求出内容大小的总和以及个数，其比值即内容大小的平均值。
    // 最后调用动作函数 min() 和 max() 得到内容大小的最小值和最大值。
    /*方法二*/
    val test01 = df.select('size cast (IntegerType))
      .agg(max('size), min('size), avg('size))
    test01.show()

    /*方法一*/
    // 2.统计不同的状态代码出现的次数。
    // 与上一步类似，首先调用变换函数 map() 取出字段 response_code ，不同的是，这里取出的是一个二元列表，即字段 response_code 和1，使得之后能够计数。
    // 再调用变换函数 reduceByKey() 按照不同的状态代码统计出现次数，并调用函数 cache() 在内存中缓存。
    // 最后调用动作函数 take() 输出成列表。
    /*方法二*/
    val test02 = df.select('response_code)
      .groupBy('response_code)
      .agg(count('response_code))
    test02.show()

    /*方式一*/
    // 3.这一步需要查看频繁访问（大于10次）的客户端主机。
    // 与上一步类似，首先调用变换函数 map() 取出包含字段 host 和1的二元列表。再调用变换函数 reduceByKey() 按照不同的主机地址统计出现次数。
    // 然后调用变换函数 filter() 取出出现次数大于10的主机地址。
    // 最后调用动作函数 take() 输出成列表。
    /*方式二*/
    val test03 = df.select('host)
      .groupBy('host)
      .agg(count('host) as "count")
      .orderBy('count desc)
      .where('count > 10)
    test03.show()

    /*方式一*/
    // 4.查看被访问最多的资源标识符。
    // 与上一步类似，首先调用变换函数 map() 取出包含字段 endpoint 和1的二元组。
    // 再调用变换函数 reduceByKey() 按照不同的资源标识符统计出现次数。
    // 最后调用动作函数 takeOrdered() 取出前10被访问最多的资源标识符。
    /*方式二*/
    val test04 = df.select('endpoint)
      .groupBy('endpoint)
      .agg(count('endpoint) as "count")
      .orderBy('count desc)
      .limit(10)
    test04.show()

    /*方式一*/
    // 5.查看前10个出现错误（即状态代码不是 200 ）最多的资源标识符（即字段 response_code ），最终的返回结果是按访问次数倒序排列的前10个标识符及其相应的访问次数。
    /*方式二*/
    val test05 = df.select('response_code cast (IntegerType))
      .where('response_code !== 200)
      .groupBy('response_code)
      .agg(count('response_code) as "count")
      .orderBy('count desc)
      .limit(10)
    test05.show()

    /*方式一*/
    // 6.统计独立主机数 统计独立主机地址（字段 host）数。
    /*方式二*/
    val test06 = df.dropDuplicates(Seq("host"))
      .count()
    println("独立主机数:" + test06)

  }

  def parseDate(str: String): String = {
    // 定义map匹配
    val month_map = Map("Jan" -> "01", "Feb" -> "02", "Mar" -> "03", "Apr" -> "04", "May" -> "05", "Jun" -> "06", "Jul" -> "07", "Aug" -> "08", "Sep" -> "09", "Oct" -> "10", "Nov" -> "11", "Dec" -> "12")
    // 01/Feb/1995:00:00:01 -0400
    val timePart = str.split(" ")(0) // 01/Feb/1995:00:00:01
    val year = timePart.substring(7, 11)
    val month = month_map.get(timePart.substring(3, 6)).getOrElse()
    val day = timePart.substring(0, 2)
    val hour = timePart.substring(12, 14)
    val minutes = timePart.substring(15, 17)
    val second = timePart.substring(18, 20)
    val date_str = year + "-" + month + "-" + day + " " + hour + ":" + minutes + ":" + second
    date_str
  }
}

